package io.iot.pipeline;

import io.iot.pipeline.decoder.OLikeProtocolDecoder;
import io.iot.pipeline.model.OLikeProtocol;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;


public class IotPipelineApplication {
    private static Log logger = LogFactory.getLog(IotPipelineApplication.class);

    public static void main(String[] args) {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup(10);

        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel channel) throws Exception {
                            channel.pipeline().addLast(
                                    createDelimiterBasedFrameDecoder(),
                                    new OLikeProtocolDecoder(),
                                    createChannelHandler()
                            );
                        }
                    })
                    .option(ChannelOption.SO_BACKLOG, 128)
                    .childOption(ChannelOption.SO_KEEPALIVE, true);

            ChannelFuture f = b.bind(7000).sync();

            f.channel().closeFuture().sync();
        } catch (Exception e) {
            logger.error(e);
        } finally {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }

    /**
     * 一般自定义ChannelHandlerAdapter处理Decode后的数据
     * @return
     */
    protected static ChannelHandler createChannelHandler(){
        return new SimpleChannelInboundHandler<OLikeProtocol>() {
            protected void channelRead0(ChannelHandlerContext ctx, OLikeProtocol msg) throws Exception {
                System.out.println(msg.toString());
            }
        };
    }

    /**
     * 粘包处理，头尾主要以`0x7E`进行判别。
     * @return
     */
    protected static DelimiterBasedFrameDecoder createDelimiterBasedFrameDecoder(){
        ByteBuf delimiter =  Unpooled.wrappedBuffer(new byte[]{0x7E});
        return new DelimiterBasedFrameDecoder(1024,true,delimiter);
    }
}
